package org.jeecg.config.binlog;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;

import java.io.IOException;
import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.*;

import static com.github.shyiko.mysql.binlog.event.EventType.*;
import static org.jeecg.config.MysqlBinLogClient.getdbTable;
import static org.jeecg.config.binlog.BinLogUtils.getColMap;

/**
 * @author gbstart
 * @create 2024-06-04 21:30
 */
@Slf4j
public class MysqlBinLogListener implements BinaryLogClient.EventListener{

    private int consumerThreads = BinLogConstants.consumerThreads;

    private BinaryLogClient parseClient;

    private BlockingQueue<BinLogItem> queue;

    private final ExecutorService consumer;

    // 存放每张数据表对应的listener
    private Multimap<String, BinLogListener> listeners;

    private Map<String, Map<String, Colum>> dbTableCols;

    private Map<Long, String> dbTableIdCols;

    /**
     * 监听器初始化
     *
     * @param conf
     */
    public MysqlBinLogListener(BinLogConstants conf) {
       // jdbc:mysql://jlmingjun.mysql.polardb.rds.aliyuncs.com:3306/tsl_db?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai
        String url = conf.getUrl();
        String substring = url.substring(url.lastIndexOf("//")+2);
        String host = substring.substring(0,substring.indexOf(":"));
        String port = substring.substring(substring.indexOf(":")+1,substring.indexOf("/"));
        System.out.println(host);
        BinaryLogClient client = new BinaryLogClient(host, Integer.parseInt(port) , conf.getUsername(), conf.getPasswd());

        EventDeserializer eventDeserializer = new EventDeserializer();
        //序列化
        eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
        );
        client.setEventDeserializer(eventDeserializer);
        this.parseClient = client;
        this.queue = new ArrayBlockingQueue<>(1024);
        this.listeners = ArrayListMultimap.create();
        this.dbTableCols = new ConcurrentHashMap<>();
        this.dbTableIdCols = new ConcurrentHashMap<>();
//自定义线程名
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-BinLogListener-runner-%d").build();
        this.consumer = new ThreadPoolExecutor(consumerThreads, consumerThreads,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), namedThreadFactory);
    }



    @Override
    public void onEvent(Event event) {
        EventType eventType = event.getHeader().getEventType();

        if (eventType == EventType.TABLE_MAP) {
            TableMapEventData tableData = event.getData();
            String db = tableData.getDatabase();
            String table = tableData.getTable();
            String dbTable = getdbTable(db, table);
            if (dbTableCols.containsKey(dbTable)) {
                dbTableIdCols.put(tableData.getTableId(), dbTable);
            }

        }


        // 只处理添加删除更新三种操作
        if (isWrite(eventType) || isUpdate(eventType) || isDelete(eventType)) {
            if (isWrite(eventType)) {
                WriteRowsEventData data = event.getData();
                for (Serializable[] row : data.getRows()) {
                    if (dbTableIdCols.containsKey(data.getTableId())) {
                        String dbTable = dbTableIdCols.get(data.getTableId());
                        BinLogItem item = BinLogItem.itemFromInsertOrDeleted(row, dbTable, dbTableCols.get(dbTable), eventType);
                        item.setDbTable(dbTable);
                        queue.add(item);
                    }
                }
            }
            if (isUpdate(eventType)) {
                UpdateRowsEventData data = event.getData();
                for (Map.Entry<Serializable[], Serializable[]> row : data.getRows()) {
                    if (dbTableIdCols.containsKey(data.getTableId())) {
                        String dbTable = dbTableIdCols.get(data.getTableId());
                        BinLogItem item = BinLogItem.itemFromUpdate(row, dbTableCols.get(dbTable), eventType);
                        item.setDbTable(dbTable);
                        queue.add(item);
                    }
                }

            }
            if (isDelete(eventType)) {
                DeleteRowsEventData data = event.getData();
                for (Serializable[] row : data.getRows()) {
                    if (dbTableIdCols.containsKey(data.getTableId())) {
                        String dbTable = dbTableIdCols.get(data.getTableId());
                        BinLogItem item = BinLogItem.itemFromInsertOrDeleted(row,dbTable, dbTableCols.get(dbTable), eventType);
                        item.setDbTable(dbTable);
                        queue.add(item);
                    }
                }
            }
        }
    }

    /**
     * 注册监听
     *
     * @param db       数据库
     * @param table    操作表
     * @param listener 监听器
     */
    public void regListener(String db, String table, BinLogListener listener) {
        String dbTable = getdbTable(db, table);
        // 获取字段集合
        Map<String, Colum> cols = getColMap(db, table);
        // 保存字段信息
        dbTableCols.put(dbTable, cols);
        // 保存当前注册的listener
        listeners.put(dbTable, listener);
    }

    /**
     * 开启多线程消费
     *
     * @throws IOException
     */
    @Async
    public void parse() throws IOException, TimeoutException {
        parseClient.registerEventListener(this);
        //可以自定义读取文件和位置
        //  parseClient.setBinlogFilename(xxx);
        //  parseClient.setBinlogPosition(xxxx);
        for (int i = 0; i < consumerThreads; i++) {
            consumer.execute(() -> {
                while (true) {
                    if (queue.size() > 0) {
                        try {
                            BinLogItem item = queue.take();
                            String dbtable = item.getDbTable();
                            listeners.get(dbtable).forEach(binLogListener -> binLogListener.onEvent(item));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    try {
                        Thread.sleep(BinLogConstants.queueSleep);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
        }
        //不设置时间将会使用主线程,没啥影响就是启动时一直转圈
        parseClient.connect(BinLogConstants.queueSleep);
        //可以设置定时器将现在的文件和位置记录下来,配合上面的代码
//Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(() -> {
//                log.info("Binlog name:{},position:{}", //parseClient.getBinlogFilename(), parseClient.getBinlogPosition());
//            }
//        }, 1, 1, TimeUnit.MINUTES);
    }

}
